package com.ds.common.util;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.function.Supplier;

/**
 * @class: FutureDemo
 * @description:
 * @author: fs
 * @create: 2023/11/22 16:58
 **/

@RequiredArgsConstructor
@Component
@Slf4j
public class FutureDemo {

    private final ThreadPoolTaskExecutor executor;


    /*单任务*/

    /**
     * runAsync无返回值
     */
    public void method1() {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            log.info("当前线程" + Thread.currentThread().getId());
            int i = 10 / 2;
            log.info("运行结果：" + i);
        }, executor);

        future.join();
    }

    /**
     * supplyAsync有返回值
     * whenComplete能感知异常，能感知结果，但没办法给返回值
     * exceptionally能感知异常，不能感知结果，能给返回值。相当于，如果出现异常就返回这个值
     */
    public void method2() {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            log.info("当前线程" + Thread.currentThread().getId());
            int i = 10 / 0;
            log.info("运行结果：" + i);
            return i;
        }, executor).whenComplete((res, excption) -> {
            //whenComplete虽然能得到异常信息，但是没办法修改返回值
            log.info("异步任务成功完成...结果是：" + res + ";异常是：" + excption);
        }).exceptionally(throwable -> {
            //exceptionally能感知异常，而且能返回一个默认值，相当于，如果出现异常就返回这个值
            return 10;
        });
        future.join();
    }

    /**
     * supplyAsync有返回值
     * handle能拿到返回结果，也能得到异常信息，也能修改返回值
     */
    public void method3() {

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            log.info("当前线程" + Thread.currentThread().getId());
            int i = 10 / 0;
            log.info("运行结果：" + i);
            return i;
        }, executor).handle((res, excption) -> {
            if (excption != null) {
                return 0;
            } else {
                return res * 2;
            }
        });
        log.info("最后结果：" + future.join());
    }

    /*单任务*/


    /*两个任务*/

    /**
     * thenRunXXX 不能接收上一次的执行结果，也没返回值
     * .thenRunAsync(() -> {
     * log.info("任务2启动了...");
     * }, executor);
     */
    public void method4() {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            log.info("当前线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("运行结果：" + i);
            return i;
        }, executor).thenRunAsync(() -> {
            log.info("任务2启动了...");
        }, executor);
        log.info("最后结果：" + future.join());
    }

    /**
     * thenAcceptXXX 能接收上一次的执行结果，但没返回值
     * .thenAcceptAsync(res->{
     * log.info("任务2启动了..."+res);
     * },executor);
     */
    public void method5() {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            log.info("当前线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("运行结果：" + i);
            return i;
        }, executor).thenAcceptAsync(res -> {
            log.info("任务2启动了..." + res);
        }, executor);

        log.info("最后结果：" + future.join());
    }

    /**
     * thenApplyXXX 能接收上一次的执行结果，又可以有返回值
     * .thenApplyAsync(res -> {
     * log.info("任务2启动了..." + res);
     * return "hello " + res;
     * }, executor);
     */
    public void method6() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            log.info("当前线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("运行结果：" + i);
            return i;
        }, executor).thenApplyAsync(res -> {
            log.info("任务2启动了..." + res);
            return "hello " + res;
        }, executor);


        log.info("最后结果：" + future.join());
    }

    /*两个任务*/


    /*三个任务*/

    /**
     * runAfterBothAsync：任务01 任务02都完成了，再开始执行任务3，不感知任务1、2的结果的，也没返回值
     */
    public void method7() {
        CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
            log.info("任务1线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("任务1结束：");
            return i;
        }, executor);
        CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
            log.info("任务2线程" + Thread.currentThread().getId());
            try {
                Thread.sleep(3000);
                log.info("任务2结束：");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);
        CompletableFuture<Void> future = future01.runAfterBothAsync(future02, () -> {
            log.info("任务3开始");
        }, executor);
        log.info("最后结果：" + future.join());
    }


    /**
     * thenAcceptBothAsync：任务01 任务02都完成了，再开始执行任务3，能感知到任务1、2的结果，但没返回值
     */
    public void method8() {
        CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
            log.info("任务1线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("任务1结束：");
            return i;
        }, executor);
        CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
            log.info("任务2线程" + Thread.currentThread().getId());
            try {
                Thread.sleep(3000);
                log.info("任务2结束：");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);
        CompletableFuture<Void> future = future01.thenAcceptBothAsync(future02, (f1, f2) -> {
            log.info("任务3开始...得到之前的结果：f1:" + f1 + ", f2:" + f2);
        }, executor);
        log.info("最后结果：" + future.join());
    }


    /**
     * thenCombineAsync：任务01 任务02都完成了，再开始执行任务3，能感知到任务1、2的结果，而且自己可以带返回值
     */
    public void method9() {
        CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
            log.info("任务1线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("任务1结束：");
            return i;
        }, executor);
        CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
            log.info("任务2线程" + Thread.currentThread().getId());
            try {
                Thread.sleep(3000);
                log.info("任务2结束：");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);
        CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> {
            return f1 + ":" + f2 + "：哈哈";
        }, executor);
        log.info("最后结果：" + future.join());
    }

    /**
     * runAfterEitherAsync：两个任务只要有一个完成，就执行任务3，不感知结果，自己没返回值
     */
    public void method10() {
        CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
            log.info("任务1线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("任务1结束：");
            return i;
        }, executor);
        CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
            log.info("任务2线程" + Thread.currentThread().getId());
            try {
                Thread.sleep(3000);
                log.info("任务2结束：");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);
        CompletableFuture<Void> future = future01.runAfterEitherAsync(future02, () -> {
            log.info("任务3开始...");
        }, executor);

        log.info("最后结果：" + future.join());
    }

    /**
     * applyToEitherAsync：两个任务只要有一个完成，就执行任务3，感知结果，自己有返回值
     */
    public void method11() {
        CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
            log.info("任务1线程" + Thread.currentThread().getId());
            int i = 10 / 4;
            log.info("任务1结束：");
            return i;
        }, executor);
        CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
            log.info("任务2线程" + Thread.currentThread().getId());
            try {
                Thread.sleep(3000);
                log.info("任务2结束：");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);
        CompletableFuture<Void> future = future01.acceptEitherAsync(future02, (res) -> {
            log.info("任务3开始...之前的结果" + res);
        }, executor);
        log.info("最后结果：" + future.join());
    }


    /*三个任务*/


    /*多任务的编排*/

    /**
     * 多任务组合
     */
    public void method12() {

        CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
            log.info("查询商品图片信息");
            return "hello.jpg";
        }, executor);

        CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
            log.info("查询商品属性信息");
            return "黑色+256G";
        }, executor);

        CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
                log.info("查询商品介绍信息");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "华为...";
        }, executor);


        CompletableFuture<Void> allOf = CompletableFuture.allOf(futureImg, futureAttr, futureDesc);
        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(futureImg, futureAttr, futureDesc);

        allOf.join();//等待所有结果完成
        anyOf.join();//等待所有结果完成

    }
    /*多任务的编排*/


    /**
     * 分批查询数据库并将结果组合起来
     */
    public void method13() {
        //List<CompletableFuture<List<SysMenu>>> futures = Lists.partition(articleIds, 250).stream()
        //        .map(o -> supplyAsyncWithDataSource(() -> mArticleDataMapper.selectArticleIdAndInteractByArticleIdsOrderInteract(o), executor)).toList();
        //List<MArticleDataVo> result = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
        //        .thenApply(v -> futures.stream().flatMap(completableFuture -> completableFuture.join().stream()).collect(Collectors.toList())).join();
    }


    //public Map<String, UrlInfo> getUrlContentMap(String content) {
    //
    //    if (StrUtil.isBlank(content)) {
    //        return new HashMap<>();
    //    }
    //    List<String> matchList = ReUtil.findAll(PATTERN, content, 0);
    //
    //    //并行请求
    //    List<CompletableFuture<Pair<String, UrlInfo>>> futures = matchList.stream().map(match -> CompletableFuture.supplyAsync(() -> {
    //        UrlInfo urlInfo = getContent(match);
    //        return Objects.isNull(urlInfo) ? null : Pair.of(match, urlInfo);
    //    })).collect(Collectors.toList());
    //    CompletableFuture<List<Pair<String, UrlInfo>>> future = FutureUtils.sequenceNonNull(futures);
    //    //结果组装
    //    return future.join().stream().collect(Collectors.toMap(Pair::getFirst, Pair::getSecond, (a, b) -> a));
    //}


    // 线程池设置多数据源
    public <T> CompletableFuture<T> supplyAsyncWithDataSource(Supplier<T> supplier, Executor executor) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                //DynamicDataSourceContextHolder.push("market-data");
                return supplier.get();
            } finally {
                //  DynamicDataSourceContextHolder.poll();
            }
        }, executor);
    }

    /*
    private SkuItemVo item(Long skuId) {
        SkuItemVo skuItemVo = new SkuItemVo();

        */

    /**
     * 3、4、5需要依赖1的运行结果，需要返回skuInfo后从中获取spuId和catalogId
     * 而2不需要依赖1的运行结果
     *//*

        //1、sku详细信息 sku_info
        CompletableFuture<SkuInfoEntity> infoFuture = CompletableFuture.supplyAsync(() -> {
            SkuInfoEntity skuInfo = getById(skuId);
            skuItemVo.setInfo(skuInfo);
            return skuInfo;
        }, executor);

        //2、sku 图片信息 sku_img  2不需要等待上边1的执行结果
        CompletableFuture<Void> imageFuture = CompletableFuture.runAsync(() -> {
            List<SkuImagesEntity> images = skuImagesService.getImagesBySkuId(skuId);
            skuItemVo.setImages(images);
        }, executor);

        //下边的3、4、5都需要上边1的执行结果
        //所以下边的3、4、5都是基于上边1的执行结果 infoFuture 开始的
        //都是以infoFuture.thenAcceptAsync(skuInfo -> {})开始的
        CompletableFuture<Void> saleAttrFuture = infoFuture.thenAcceptAsync(skuInfo -> {
            //3、spu 销售属性组合  3
            List<SkuItemSaleAttrVo> saleAttr = skuSaleAttrValueService.getSaleAttrBySpuId(skuInfo.getSpuId());
            skuItemVo.setSaleAttr(saleAttr);
            log.info(saleAttr);
        }, executor);

        CompletableFuture<Void> descFuture = infoFuture.thenAcceptAsync(skuInfo -> {
            //4、spu 的介绍
            SpuInfoDescEntity spuInfoDesc = spuInfoDescService.getById(skuInfo.getSpuId());
            skuItemVo.setDesc(spuInfoDesc);
        }, executor);

        CompletableFuture<Void> attrGroupFuture = infoFuture.thenAcceptAsync(skuInfo -> {
            //5、spu 规格参数信息
            List<SpuItemAttrGroupVo> groupAttrs = attrGroupService.getAttrGroupWithAttrsBySpuId(skuInfo.getSpuId(), skuInfo.getCatalogId());
            log.info(groupAttrs);
            skuItemVo.setGroupAttrs(groupAttrs);
        }, executor);


        //等待所有任务完成
        try {
            CompletableFuture.allOf(saleAttrFuture, descFuture, attrGroupFuture, imageFuture).get();
        } catch (InterruptedException e) {
            log.error("查询商品详情异步编排错误: ");
            log.error(e.getMessage());
        } catch (ExecutionException e) {
            log.error(e.getMessage());
        }
        return skuItemVo;
    }
*/

    /**
     * 并行查询redis
     *
     * @return
     */
    public List<Map<String, Object>> method() {
        List<String> list = new ArrayList<>();
        List<CompletableFuture<Map<String, Object>>> futures = list.stream().map(streamName -> CompletableFuture.supplyAsync(() -> {
            String fullStreamName = "STREAM_" + streamName;
            String groupName = streamName + "_GROUP";
            //PendingMessagesSummary pending = redisTemplate.opsForStream().pending(fullStreamName, groupName);
            PendingMessagesSummary pending = null;
            if (Objects.isNull(pending)) {
                return null;
            }
            Map<String, Object> res = new HashMap<>();
            res.put("streamName", fullStreamName);
            res.put("groupName", groupName);
            res.put("pendingMessages", pending.getTotalPendingMessages());
            log.info("streamName:{}, groupName:{}, pendingMessages:{}", fullStreamName, groupName, pending.getTotalPendingMessages());
            return res;
        })).toList();
        CompletableFuture<List<Map<String, Object>>> future = FutureUtils.sequenceNonNull(futures);
        return future.join().stream().toList();
    }


}
